"""
Case Type   : 子事务并发回滚功能模块-5.1.0新增特性
Case Name   : 带子事务的多个事务并发进行rollback，事务包含部分子事务进行了release或者rollback
Create At   : 2024/1/4
Owner       : peilinqian
Description :
    1. 创建表
    2. 创建多个回滚存储过程，包含部分子事务进行了release或者rollback
    3. 初始查询ProcArrayLock锁的wait情况
    4. 并发执行存储过程
    5. 再次查询ProcArrayLock锁的wait情况
    6. 最终验证并发事务执行结果
Expect      :
    1. 创建表，成功
    2. 创建多个回滚存储过程，包含部分子事务进行了release或者rollback，成功
    3. 初始查询ProcArrayLock锁的wait情况，正常
    4. 并发执行存储过程，成功
    5. 再次查询ProcArrayLock锁的wait情况，较步骤3增加
    6. 最终验证并发事务执行结果，结果正确
History     :
"""
import os
import unittest

from testcase.utils.ComThread import ComThread
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger


class GscTestCase(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)}:start----')
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.com = Common()
        self.t_name = 't_subxact_case0004'
        self.proc_name1 = 'proc_subxact_case0004_1'
        self.proc_name2 = 'proc_subxact_case0004_2'
        self.proc_name3 = 'proc_subxact_case0004_3'
        self.thread_num = 10

    def test_main(self):
        step_txt = '----step1:创建表; expect:创建成功----'
        self.log.info(step_txt)
        sql = f"drop table if exists {self.t_name};" \
            f"create table {self.t_name}(col1 int);"
        create_result = self.pri_sh.execut_db_sql(sql)
        self.log.info(create_result)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, create_result,
                      '执行失败:' + step_txt)

        step_txt = '----step2:创建多个回滚存储过程，包含部分子事务进行了release或者rollback; ' \
                   'expect:创建成功----'
        self.log.info(step_txt)
        sql = f'''
            create or replace procedure {self.proc_name1}()
            as
            declare
            v_str text;
            begin
              for i in 1..50000 loop
                if i % 10 = 0 then
                  v_str := concat('savepoint', i);
                  insert into {self.t_name}(col1) values (i);
                  savepoint v_str;
                else
                  insert into {self.t_name}(col1) values (i);
                end if;
                if i = 50000 then
                  rollback;
                end if;
              end loop;
            end;
        '''
        create_result = self.pri_sh.execut_db_sql(sql)
        self.log.info(create_result)
        self.assertIn(self.constant.CREATE_PROCEDURE_SUCCESS_MSG,
                      create_result, '执行失败:' + step_txt)
        sql = f'''
            create or replace procedure {self.proc_name2}()
            as
            declare
            v_str text;
            begin
              for i in 1..50000 loop
                if i % 10 = 0 then
                  v_str := concat('savepoint', i);
                  savepoint v_str;
                  insert into {self.t_name}(col1) values (i);
                  rollback to v_str;
                else 
                  insert into {self.t_name}(col1) values (i);
                end if;
                if i = 50000 then
                  rollback;
                end if;
              end loop;
            end;
        '''
        create_result = self.pri_sh.execut_db_sql(sql)
        self.log.info(create_result)
        self.assertIn(self.constant.CREATE_PROCEDURE_SUCCESS_MSG,
                      create_result, '执行失败:' + step_txt)
        sql = f'''
            create or replace procedure {self.proc_name3}()
            as
            declare
            v_str text;
            begin
              for i in 1..50000 loop
                v_str := concat('savepoint', i);
                if i % 10 = 0 then
                  savepoint v_str;
                  insert into {self.t_name}(col1) values (i);
                  release savepoint v_str;
                else 
                  insert into {self.t_name}(col1) values (i);
                end if;
                if i % 20 = 0 then
                  savepoint v_str;
                end if;
                if i = 50000 then
                  rollback;
                end if;
              end loop;
            end;
        '''
        create_result = self.pri_sh.execut_db_sql(sql)
        self.log.info(create_result)
        self.assertIn(self.constant.CREATE_PROCEDURE_SUCCESS_MSG,
                      create_result, '执行失败:' + step_txt)

        step_txt = '----step3:初始查询ProcArrayLock锁的wait情况; expect:查询结果正常----'
        self.log.info(step_txt)
        check_sql = f"select * from dbe_perf.wait_events where " \
            f"event='ProcArrayLock';"
        select_result = self.pri_sh.execut_db_sql(check_sql)
        self.log.info(select_result)
        wait_res = self.com.parse_sql_res_with_col_name(select_result, False)
        wait_1 = int(wait_res[0]['wait'])
        self.assertGreaterEqual(wait_1, 0, '执行失败:' + step_txt)

        step_txt = '----step4:并发执行存储过程; expect:执行正常----'
        self.log.info(step_txt)
        call_thread = []
        call_result = []
        call_sql_1 = f"call {self.proc_name1}();"
        call_sql_2 = f"call {self.proc_name2}();"
        call_sql_3 = f"call {self.proc_name3}();"
        for j in range(self.thread_num):
            call_thread.append(
                ComThread(self.pri_sh.execut_db_sql, args=(call_sql_1,)))
            call_thread.append(
                ComThread(self.pri_sh.execut_db_sql, args=(call_sql_2,)))
            call_thread.append(
                ComThread(self.pri_sh.execut_db_sql, args=(call_sql_3,)))
        for j in range(self.thread_num * 3):
            call_thread[j].setDaemon(True)
            call_thread[j].start()
        for j in range(self.thread_num * 3):
            call_thread[j].join(50)
            result = call_thread[j].get_result()
            call_result.append(result)
        if len(set(call_result)) != 3:
            self.log.info(call_result)
        self.assertEqual(len(set(call_result)), 3, '执行失败:' + step_txt)
        self.assertEqual(len(call_result), self.thread_num * 3,
                         '执行失败:' + step_txt)

        step_txt = '----step5:再次查询ProcArrayLock锁的wait情况; expect:较步骤3增加----'
        self.log.info(step_txt)
        select_result = self.pri_sh.execut_db_sql(check_sql)
        self.log.info(select_result)
        wait_res = self.com.parse_sql_res_with_col_name(select_result, False)
        wait_2 = int(wait_res[0]['wait'])
        self.assertGreaterEqual(wait_2, wait_1, '执行失败:' + step_txt)

        step_txt = '----step6:最终验证并发事务执行结果; expect:结果正确----'
        self.log.info(step_txt)
        sql = f"select count(*) from {self.t_name};"
        select_result = self.pri_sh.execut_db_sql(sql)
        self.log.info(select_result)
        count_result = int(select_result.splitlines()[-2].strip())
        self.assertEqual(count_result, 0, '执行失败:' + step_txt)

    def tearDown(self):
        step_txt = '----this is teardown----'
        self.log.info(step_txt)
        step1_txt = '----删除表、存储过程; expect:删除成功----'
        self.log.info(step1_txt)
        sql = f"drop table if exists {self.t_name};" \
            f"drop procedure if exists {self.proc_name1};" \
            f"drop procedure if exists {self.proc_name2};" \
            f"drop procedure if exists {self.proc_name3};"
        drop_result = self.pri_sh.execut_db_sql(sql)
        self.log.info(drop_result)

        self.log.info(f'----{os.path.basename(__file__)}:end----')
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, drop_result,
                      '执行失败:' + step_txt)
        self.assertIn(self.constant.DROP_PROCEDURE_SUCCESS_MSG, drop_result,
                      '执行失败:' + step_txt)
